package com.bicon.botu.core.base.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;

import java.lang.reflect.Field;
import java.util.Map;

/**
 * <p>
 * kafka 生产者监听器
 * </p>
 *
 * @author tao.zeng.
 * @since 2018/1/15.
 */
@Slf4j
public class KafkaProducerListener implements ProducerListener<String, String> {


    private KafkaProducerCallbackListener kafkaProducerCallbackListener;

    public KafkaProducerListener() {

    }

    /**
     * 反射获取kafka服务器地址
     *
     * @param kafkaTemplate
     */
    public KafkaProducerListener(KafkaTemplate kafkaTemplate) {
        getBootstrapServers(kafkaTemplate);
    }

    /**
     * 设置回调监听器
     *
     * @param kafkaProducerCallbackListener
     */
    public void setKafkaProducerCallbackListener(KafkaProducerCallbackListener kafkaProducerCallbackListener) {
        this.kafkaProducerCallbackListener = kafkaProducerCallbackListener;
    }

    /**
     * 获取 kafka 服务器地址以及端口
     *
     * @param kafkaTemplate kafka 操作对象
     */
    public Map getBootstrapServers(KafkaTemplate kafkaTemplate) {
        try {
            Class<? extends KafkaTemplate> clz = kafkaTemplate.getClass();
            Field field = clz.getDeclaredField("producerFactory");
            field.setAccessible(true);
            DefaultKafkaProducerFactory producerFactory = (DefaultKafkaProducerFactory) field.get(kafkaTemplate);
            Class<? extends DefaultKafkaProducerFactory> cls = producerFactory.getClass();
            Field configs = cls.getDeclaredField("configs");
            configs.setAccessible(true);
            Map map = (Map) configs.get(producerFactory);
            log.info("kafka 服务器地址为:{}", map.get("bootstrap.servers"));
            return map;
        } catch (Exception e) {
            log.error("反射获取kafka地址失败，{}", e);
        }
        return null;
    }

    /**
     * 发送消息成功
     *
     * @param topic          topic
     * @param partition      分区
     * @param key            key
     * @param value          value
     * @param recordMetadata 元数据
     */
    @Override
    public void onSuccess(String topic, Integer partition, String key, String value, RecordMetadata recordMetadata) {

        if (kafkaProducerCallbackListener != null) {
            kafkaProducerCallbackListener.onSuccess(topic, value);
        }

        log.info("==========kafka发送数据成功（日志开始）==========");
        log.info("----------topic:{}", topic);
        log.info("----------partition:{}", partition);
        log.info("----------key:{}", key);
        log.info("----------value:{}", value);
        log.info("----------RecordMetadata:{}", recordMetadata);
        log.info("==========kafka发送数据成功（日志结束）==========");
    }

    /**
     * 发送消息失败
     *
     * @param topic
     * @param partition 分区
     * @param key
     * @param value
     * @param exception
     */
    @Override
    public void onError(String topic, Integer partition, String key, String value, Exception exception) {

        if (kafkaProducerCallbackListener != null) {
            kafkaProducerCallbackListener.onFailure(topic, value, exception);
        }

        log.error("==========kafka发送数据错误（日志开始）==========");
        log.warn("----------topic:{}", topic);
        log.warn("----------partition:{}", partition);
        log.warn("----------key:{}", key);
        log.warn("----------value:{}", value);
        log.warn("----------Exception:{}", exception);
        log.warn("==========kafka发送数据错误（日志结束）==========");
    }

    /**
     * 方法返回值代表是否启动kafkaProducer监听器
     */
    @Override
    public boolean isInterestedInSuccess() {
        log.info("///kafkaProducer监听器启动///");
        return true;
    }

    /**
     * kafka 回调监听器
     */
    public interface KafkaProducerCallbackListener {
        /**
         * 发送成功
         *
         * @param topic topic
         * @param value 发送的数据
         */
        void onSuccess(String topic, String value);

        /**
         * 发送失败
         *
         * @param topic     topic
         * @param value     发送的数据
         * @param exception 异常
         */
        void onFailure(String topic, String value, Exception exception);
    }
}
